# -*- coding: utf-8 -*-
# @Time      : 2025-08-01
# @Author    : 2023090007@ahdy.edu.cn
# @File      : master
# @Desc  
import errno
import gc
import multiprocessing
import os
import signal
import time
import traceback
from service import Service
from worker import Worker
from common.loghandler import logger


class Master(Service):
    """
    核心层
    """

    def __init__(self):
        super().__init__()
        self.worker_process = []
        self.running = True
        self.pause_wait_and_respawn_workers = False

    
    def init_worker_process(self):
        for n in range(self.worker_process_num):
            worker = Worker()
            worker.no = n  # 进程序号
            process = multiprocessing.Process(target=worker)
            process.daemon = True  # 父进程正常退出(非os._exit) 会同时关闭子进程的
            worker.process = process

            process.start()
            worker.pid = process.pid  # 只有启动只有才会有pid

            self.worker_process.append(worker)

    def start(self):
        """
        create:0
        submit:1
        success:10
        :return:
        """
        logger.info("master start ...")
        self.initialize()
        self.signal()
        self.init_worker_process()
        # wait until the condition is set by stop()
        while True:
            try:
              
                time.sleep(1)
                if not self.running:
                    break
                if self.pause_wait_and_respawn_workers:
                    continue
                # worker进程数量维护
                self.wait_and_respawn_workers() 
            except (SystemExit, KeyboardInterrupt):
                logger.warn("Master:SystemExit, KeyboardInterrupt")
                break
            except Exception as ex:
                logger.error("exception:%s", ex)
                traceback.print_exc()
        # SIGKILL 只能运行在Linux系统
        for p in multiprocessing.active_children():
            os.kill(p.pid, signal.SIGKILL)  # 进程强制终止-9 且不能捕获此信号
            p.join()  # 子进程
        os._exit(0)

            
    def wait_and_respawn_workers(self):
        """
        监听僵死子进程
        :return:
        """
        try:
            while True:
                cpid, status = os.waitpid(-1, os.WNOHANG)  # 获取退出子进程的pid和状态，没有则立即返回
                if cpid == 0:
                    logger.info("All child process healthy.")
                    break  # No child process was immediately available.
                if os.WIFEXITED(status):
                    # 正常退出
                    exitcode = os.WEXITSTATUS(status)
                    logger.info('worker with pid %s exit with exitcode %s, starting new worker...', cpid, exitcode)
                else:
                    # 信号退出
                    pass
                    sig = os.WIFSIGNALED(status)
                    logger.info('worker with pid %s exit with sig %s, starting new worker...', cpid, sig)
                for idx in range(len(self.worker_process)):
                    worker = self.worker_process[idx]
                    if worker.pid == cpid:  # 重新创建并启动一个子进程，保证子进程数量恒定
                        w = multiprocessing.Process(target=worker)
                        w.daemon = True
                        worker.process = w
                        w.start()
                        worker.pid = w.pid  # 启动后赋值 否则是None
        except OSError as e:
            logger.warn("The wait_and_respawn_workers OSError:%s", e)
            if e.errno == errno.ECHILD:
                pass  # Current process has no existing unwaited-for child processes.
            else:
                raise
        finally:
            gc.collect()

    def signal(self):
        """
        设置信号处理方法
        注意:信号处理方法中使用logger会导致多线程死锁,信号处理线程和主线程logger lock 导致死锁
        :return:
        """
        logger.info("start signal...")
        signal.signal(signal.SIGTERM, self.handle_sigterm) # 强行终止
        signal.signal(signal.SIGINT, self.handle_sigint) # 强行终止
        signal.signal(signal.SIGQUIT, self.handle_sigquit) # 优雅终止
        signal.signal(signal.SIGHUP, self.handle_sighup) # 重启服务
    

    def handle_sigint(self, sig, frame):
        """
        快速停止服务  kill -2 master_pid
        multiprocessing模块来创建daemon进程，父进程终止，子进程也会终止(和直接通过fork创建子进程有区别)
        :param sig:
        :param frame:
        :return:
        """
        logger.info("master pid:%s handle_sigint start ...", os.getpid())
        self.running = False # 父进程正常退出 子进程（守护进程）会强制终止

    def handle_sigterm(self, sig, frame):
        """
        快速停止服务  kill -15 master_pid
        multiprocessing模块来创建daemon进程，父进程终止，子进程也会终止(和直接通过fork创建子进程有区别)
        :param sig:
        :param frame:
        :return:
        """
        logger.info("master pid:%s handle_sigterm start ...", os.getpid())
        self.running = False # 父进程正常退出 子进程（守护进程）会强制终止
        

    def handle_sigquit(self, sig, frame):
        """
        优雅关闭服务
        kill -3 master_pid
        :param sig:
        :param frame:
        :return:
        """
        logger.info("master pid:%s handle_sigquit start ...", os.getpid())
        self.pause_wait_and_respawn_workers = True;
        for p in multiprocessing.active_children():
            os.kill(p.pid, signal.SIGQUIT)
            p.join()  # 等待子进程结束
        self.running = False  # respawn 不会重新拉起子进程
        

    def handle_sighup(self, sig, frame):
        """
        重启服务
        kill -1 master_pid
        :param sig:
        :param frame:
        :return:
        """
        logger.info("master pid:%s handle_sighup start ...", os.getpid())
        self.pause_wait_and_respawn_workers = True
        for p in multiprocessing.active_children():
            os.kill(p.pid, signal.SIGQUIT)
            p.join()  # 等待子进程处理完当前任务后正常退出
        self.parse_config() # 重新加载配置文件
        self.init_worker_process()
        self.pause_wait_and_respawn_workers = False # running=True 会重新拉起进程
